package com.huatai.saas.kafka;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.huatai.common.constant.KafkaConstant;
import com.huatai.model.dubbo.ModelDubboApi;
import com.huatai.model.entity.Model;
import com.huatai.saas.constant.KafkaTopicConstant;
import com.huatai.saas.dto.SceneStatusDto;
import com.huatai.saas.entity.SceneFileEntity;
import com.huatai.saas.entity.SceneFileHistoryEntity;
import com.huatai.saas.entity.UserModelAuth;
import com.huatai.saas.service.FileService;
import com.huatai.saas.service.SceneFileHistoryService;
import com.huatai.saas.service.SceneFileService;
import com.huatai.saas.service.UserModelAuthService;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * @program: nanming-saas
 * @description: saas监听消息
 * @author: songxiude
 * @create: 2024-06-24 10:07
 **/
@Slf4j
@Component
public class SaaSKafkaListener {
	@Autowired
	private FileService fileService;
	@Autowired
	private SceneFileService sceneFileService;
	@Autowired
	private SceneFileHistoryService sceneFileHistoryService;

	@Autowired
	private UserModelAuthService userModelAuthService;

	@DubboReference(check = false)
	private ModelDubboApi modelDubboApi;


	@KafkaListener(groupId = "htsaas_scene", topics = KafkaTopicConstant.HT_SAAS_SCENE)
	public void sceneListener(ConsumerRecord record) {
		String message = (String) record.value();
		SceneStatusDto sceneStatusDto = JSON.parseObject(message, SceneStatusDto.class);
		this.updateSceneFile(sceneStatusDto);

	}

	@KafkaListener(groupId = "htsaas_model", topics = KafkaConstant.MODEL_EDIT_INFO)
	public void modelListener(ConsumerRecord record) {
		String message = (String) record.value();
		JSONObject parseObject = JSON.parseObject(message, JSONObject.class);
		Long mdId = (Long) parseObject.get("md_id");
		int editType = (int) parseObject.get("edit_type");
		if (editType == 1) {
			Model model = modelDubboApi.getModelById(mdId);
			userModelAuthService.update(null, Wrappers.<UserModelAuth>lambdaUpdate()
				.set(UserModelAuth::getModelName, model.getModelName())
				.set(UserModelAuth::getIndustryKey, model.getIndustryKey())
				.eq(UserModelAuth::getModelId, mdId));
		} else if (editType == 2) {
			userModelAuthService.remove(Wrappers.<UserModelAuth>lambdaQuery().eq(UserModelAuth::getModelId, mdId));

		}

	}

	@Transactional(rollbackFor = Exception.class)
	private void updateSceneFile(SceneStatusDto sceneStatusDto) {
		try {
			//贺才云修改
			if (sceneStatusDto.getStatus()) {
				SceneFileEntity sceneFileEntity = sceneFileService.getById(sceneStatusDto.getBussId());
				if (ObjectUtil.isNotEmpty(sceneFileEntity)) {
					String version = sceneFileEntity.getVersion();
					LambdaQueryWrapper<SceneFileHistoryEntity> eq = Wrappers.<SceneFileHistoryEntity>lambdaQuery()
						.eq(SceneFileHistoryEntity::getSfId, sceneStatusDto.getBussId())
						.eq(SceneFileHistoryEntity::getVersion, version)
						.eq(SceneFileHistoryEntity::getPath, sceneStatusDto.getPath());
					SceneFileHistoryEntity sceneFileHistoryEntity = sceneFileHistoryService.getOne(eq);
					if (ObjectUtil.isNotEmpty(sceneFileHistoryEntity)) {
						Long fileSize = fileService.getFileSize(sceneFileHistoryEntity.getPath());
						sceneFileEntity.setTotalSize(fileSize);
						sceneFileHistoryEntity.setTotalSize(fileSize);
						sceneFileHistoryService.updateById(sceneFileHistoryEntity);
						sceneFileService.updateById(sceneFileEntity);
					}
				}
			}
		} catch (Exception e) {
			log.error("场景包上传状态监听异常", e.getMessage());
		}
	}
}
